package com.yangu.flink.cdc

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

object Flink_CDCWithSQL {

  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment

    val settings = EnvironmentSettings.newInstance()
      .inStreamingMode()
      .build()

    val tableEnvironment = StreamTableEnvironment.create(environment, settings)

    tableEnvironment.executeSql(
      """
        |CREATE TABLE user_info (
        |     db_name STRING METADATA FROM 'database_name' VIRTUAL,
        |     table_name STRING METADATA  FROM 'table_name' VIRTUAL,
        |     operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
        |     id INT,
        |     name STRING,
        |     age INT,
        |     sex STRING,
        |     create_time TIMESTAMP,
        |     PRIMARY KEY(id) NOT ENFORCED
        |     ) WITH (
        |     'connector' = 'mysql-cdc',
        |     'hostname' = '172.20.3.43',
        |     'port' = '3307',
        |     'username' = 'root',
        |     'password' = 'MySQL!23',
        |     'database-name' = 'test',
        |     'table-name' = 'userInfo')
        |""".stripMargin)

    tableEnvironment.executeSql(
      """
        |SELECT
        | *
        |FROM
        | user_info
        |""".stripMargin)
      .print()

    environment.execute()

  }

}
